/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.dataredistribution.handler;

import cn.ac.iie.ulss.dataredistribution.commons.GlobalVariables;
import cn.ac.iie.ulss.dataredistribution.commons.RuntimeEnv;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author evan
 */
public class FileOperator {

    private static ConcurrentHashMap<String, Object[]> valueToFile = (ConcurrentHashMap<String, Object[]>) RuntimeEnv.getParam(GlobalVariables.VALUE_TO_FILE);
    private static ConcurrentHashMap<Long, AtomicLong> fileToCount = (ConcurrentHashMap<Long, AtomicLong>) RuntimeEnv.getParam(GlobalVariables.FILE_TO_COUNT);
    private static int fileMaxSize = (Integer) RuntimeEnv.getParam(RuntimeEnv.FILE_MAXSIZE);
    static org.apache.log4j.Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = org.apache.log4j.Logger.getLogger(FileOperator.class.getName());
    }

    static Object[] getFileFromValueToFile(String value) {
        Object[] o = valueToFile.get(value);
        return o;
    }

    static synchronized void putFile(String value, Object[] o) {
        valueToFile.put(value, o);
        Long fid = (Long) o[1];
        AtomicLong al = new AtomicLong(0);
        fileToCount.put(fid, al);
    }

    static synchronized void removeFile(long fid) {
        String tmp = null;
        for (String t : valueToFile.keySet()) {
            Object[] o = valueToFile.get(t);
            if ((Long) o[1] == fid) {
                tmp = t;
                break;
            }
        }
        if (tmp != null) {
            valueToFile.remove(tmp);
            logger.info("valuetofile remove the file " + fid + " for the key " + tmp);
        }

        AtomicLong al = fileToCount.remove(fid);
        if (al != null) {
            logger.info("has sended " + al.longValue() + " messages to the file " + fid);
        }
    }

    static synchronized void removeFile(String value) {
        Object[] o = valueToFile.remove(value);
        if (o != null) {
            Long fid = (Long) o[1];
            logger.info("valuetofile remove the file " + fid + " for the key " + value);
            if (fid != null) {
                AtomicLong al = fileToCount.remove(fid);
                if (al != null) {
                    logger.info("has sended " + al.longValue() + " messages to the file " + fid);
                }
            }
        }
    }

    static synchronized boolean checkFile(long fid) {
        AtomicLong al = fileToCount.get(fid);
        if (al != null) {
            if (al.longValue() > fileMaxSize) {
                return false;
            } else {
                return true;
            }
        } else {
            return false;
        }
    }

    static synchronized void add(long fid, long count) {
        AtomicLong al = fileToCount.get(fid);
        if(al!=null){
            al.addAndGet(count);
        }
    }
}
